package org.tao.lightningmq.broker.netty.nameserver;

import com.alibaba.fastjson2.JSON;
import io.netty.util.internal.StringUtil;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.tao.lightningmq.broker.cache.CommonCache;
import org.tao.lightningmq.broker.config.GlobalProperties;
import org.tao.lightningmq.broker.enums.BrokerClusterModeEnum;
import org.tao.lightningmq.common.dto.PullBrokerIpDTO;
import org.tao.lightningmq.common.dto.PullBrokerIpRespDTO;
import org.tao.lightningmq.common.dto.ServiceRegistryReqDTO;
import org.tao.lightningmq.common.enums.BrokerRegistryRoleEnum;
import org.tao.lightningmq.common.enums.NameServerEventCode;
import org.tao.lightningmq.common.enums.NameServerResponseCode;
import org.tao.lightningmq.common.enums.RegistryTypeEnum;
import org.tao.lightningmq.common.remote.BrokerMasterChangeCallBack;
import org.tao.lightningmq.common.remote.NameServerNettyRemoteClient;
import org.tao.lightningmq.common.tcp.TcpMsg;

import java.util.HashMap;
import java.util.Map;
import java.util.UUID;
import java.util.concurrent.ConcurrentHashMap;

/**
 * @Author lht
 * @date 2025/1/8 - 17:10
 * @description:
 */
public class NameServerClient {

    private static final Logger LOGGER = LoggerFactory.getLogger(NameServerClient.class);

    private final Map<String, NameServerNettyRemoteClient> clientMap = new ConcurrentHashMap<>();

    public void initConnection() {
        String nameserverIpStr = CommonCache.getGlobalProperties().getNameserverIp();
        String nameserverPortStr = CommonCache.getGlobalProperties().getNameserverPort();
        if (StringUtil.isNullOrEmpty(nameserverIpStr) || StringUtil.isNullOrEmpty(nameserverPortStr)) {
            throw new RuntimeException("error port or ip");
        }
        String[] nameserverIpArr = nameserverIpStr.split(";");
        String[] nameserverPortArr = nameserverPortStr.split(";");
        if (nameserverPortArr.length != nameserverIpArr.length) {
            throw new IllegalArgumentException("nameserverIpArr和nameserverPortArr长度不对等");
        }
        for(int i = 0; i < nameserverIpArr.length; i++) {
            NameServerNettyRemoteClient nameServerNettyRemoteClient =
                    new NameServerNettyRemoteClient(nameserverIpArr[i], Integer.parseInt(nameserverPortArr[i]));
            nameServerNettyRemoteClient.buildConnection();
            clientMap.put(nameserverIpArr[i] + ":" + nameserverPortArr[i], nameServerNettyRemoteClient);
            nameServerNettyRemoteClient.setNameServerLogoutCallBack(() -> {
                nameServerNettyRemoteClient.buildConnection();
                sendRegistryMsg(nameServerNettyRemoteClient);
            });
        }
//        nameServerNettyRemoteClient = new NameServerNettyRemoteClient(nameserverIp, nameserverPort);
//        nameServerNettyRemoteClient.buildConnection();
    }

    public void sendRegistryMsg() {
        for (NameServerNettyRemoteClient value : clientMap.values()) {
            if (value.isActive()) {
                sendRegistryMsg(value);
            }
        }
    }

    public void sendRegistryMsg(NameServerNettyRemoteClient nsClient) {
        GlobalProperties globalProperties = CommonCache.getGlobalProperties();
        Map<String, Object> attrs = new HashMap<>();
        ServiceRegistryReqDTO registryDTO = new ServiceRegistryReqDTO();
        registryDTO.setIp(globalProperties.getBrokerIp());
        registryDTO.setPort(globalProperties.getBrokerPort());
        registryDTO.setUser(globalProperties.getNameserverUser());
        registryDTO.setPassword(globalProperties.getNameserverPassword());
        registryDTO.setRegistryType(RegistryTypeEnum.BROKER.getCode());
        registryDTO.setAttrs(attrs);
        registryDTO.setMsgId(UUID.randomUUID().toString());

        String clusterMode = globalProperties.getBrokerClusterMode();
        if (StringUtil.isNullOrEmpty(clusterMode)) {
            attrs.put("role", "single");
        } else if (BrokerClusterModeEnum.MASTER_SLAVE.getCode().equals(clusterMode)) {
            //注册模式是集群架构
            BrokerRegistryRoleEnum brokerRegistryEnum = BrokerRegistryRoleEnum.getEnumByCode(globalProperties.getBrokerClusterRole());
            attrs.put("role", brokerRegistryEnum.getCode());
        }
        attrs.put("group", globalProperties.getBrokerClusterGroup());

        TcpMsg tcpMsg = new TcpMsg(NameServerEventCode.REGISTRY.getCode(), JSON.toJSONBytes(registryDTO));
        TcpMsg registryResponseMsg = nsClient.sendSyncMsg(tcpMsg, registryDTO.getMsgId());
        if (registryResponseMsg == null) {
            LOGGER.warn("send registry msg to nameserver timeout");
            return;
        }
        if (registryResponseMsg.getCode() == NameServerResponseCode.REGISTRY_SUCCESS.getCode()) {
            LOGGER.info("broker registry success");
            CommonCache.getHeartBeatTaskManager().startHeartBeatTask();
        } else if (NameServerResponseCode.ERROR_USER_OR_PASSWORD.getCode() == registryResponseMsg.getCode()) {
            //验证失败，抛出异常
            throw new RuntimeException("error nameserver user or password");
        }
    }

    /**
     * 查询master broker地址
     *
     * @return master broker地址
     */
    public String queryBrokerMasterAddress() {
        NameServerNettyRemoteClient nsClient = getUsefulNSClient();
        String clusterMode = CommonCache.getGlobalProperties().getBrokerClusterMode();
        if (!BrokerClusterModeEnum.MASTER_SLAVE.getCode().equals(clusterMode)) {
            return null;
        }
        PullBrokerIpDTO pullBrokerIpDTO = new PullBrokerIpDTO();
        pullBrokerIpDTO.setBrokerClusterGroup(CommonCache.getGlobalProperties().getBrokerClusterGroup());
        pullBrokerIpDTO.setRole(BrokerRegistryRoleEnum.MASTER.getCode());
        pullBrokerIpDTO.setMsgId(UUID.randomUUID().toString());
        TcpMsg tcpMsg = new TcpMsg(NameServerEventCode.PULL_BROKER_IP_LIST.getCode(),
                JSON.toJSONBytes(pullBrokerIpDTO));
        TcpMsg resp = nsClient.sendSyncMsg(tcpMsg, pullBrokerIpDTO.getMsgId());
        PullBrokerIpRespDTO pullBrokerIpRespDTO = JSON.parseObject(resp.getBody(), PullBrokerIpRespDTO.class);
        return pullBrokerIpRespDTO.getMasterAddressList().get(0);
    }

    private NameServerNettyRemoteClient getUsefulNSClient() {
        for (NameServerNettyRemoteClient value : clientMap.values()) {
            if (value.isActive()) {
                return value;
            }
        }
        throw new RuntimeException("没有已连接的Nameserver");
    }

    public void setBrokerMasterChangeCallBack(BrokerMasterChangeCallBack changeCallBack) {
        for (NameServerNettyRemoteClient value : clientMap.values()) {
            if (value.isActive()) {
                value.setBrokerMasterChangeCallBack(changeCallBack);
            }
        }
//        this.nameServerNettyRemoteClient.setBrokerMasterChangeCallBack(changeCallBack);
    }

    public void sendHeartBeatMsg(TcpMsg tcpMsg) {
        for (NameServerNettyRemoteClient value : clientMap.values()) {
            if (value.isActive()) {
                value.sendAsyncMsg(tcpMsg);
            }
        }
    }
}
